//
// Created by root on 11/8/18.
//


#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <dirent.h>
#include <inttypes.h>
#include <stdbool.h>
#include <errno.h>
#include <time.h>
#include <fcntl.h>
#include <unistd.h>
#include <fnmatch.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/file.h>
#include <sys/time.h>
#include <sys/mman.h>

#include <attr/xattr.h>
#include <sys/file.h>
#include <sys/mman.h>

#include "log.h"
#include "backend-fs.h"
#include "utcache.h"
#include "incubator.h"
#include "crc32.h"
#include "ioengine.h"
#include "ursax-crc32.h"
#include "chunkserver-journal/chunkserver-journal.h"
//#include "raw-chunk-fs.h"

#define DIRECTORY_VERSION "0"
#define USE_SSD_FILE "disktype.ssd"
#define USE_RAW_FS_FILE "ursax.rawfs.path"

#define IS_CRC_INVOLVED_OFFSET(offset) (offset % (CRC32_BLOCK_SIZE + sizeof(uint32_t)) == 0)


int lockfd = -1;
int io_extra_flag = 0;
uint32_t o_direct_size = 512;
struct RawChunkFS* rfs = NULL;

struct foo_cache* fdcache = NULL;

int in_flight_write = 0;

st_mutex_t open_chunk_lock;

void* checker;

#define USER_VERSION "user.version"

uint32_t g_opened_chunk_nb = 0;
uint64_t g_opened_chunk_mem_use = 0;

static inline uint32_t int64hash(uint64_t x)
{
    return inthash((uint32_t)x) ^ inthash((uint32_t)(x>>32));
}

bool test_size(const char* path, int size)
{
    char tmppath[PATH_MAX];
    sprintf(tmppath, "%s/invisable", path);
    bool ret = false;
    int fd = open(tmppath, O_RDWR | O_CREAT | O_DIRECT, 0600);
    if(fd <= 0) {
        LOG_ERROR("error fd < 0 %m");
        return ret;
    }

    char *buf = (char*)memalign(SIZE_4K, size);
    int r = write(fd, buf, size);
    if(r == size){
        ret = true;
    }

    free(buf);
    close(fd);
    unlink(tmppath);
    return ret;
}

static inline void new_a_chunk()
{
    g_opened_chunk_nb++;
    g_opened_chunk_mem_use += sizeof(struct Chunk);
    LOG_INFO("new a chunk, opened chunk nb is %d opened chunk mem use is %ld",
            g_opened_chunk_nb, g_opened_chunk_mem_use);
}

void be_chunk_start_use_journal(struct Chunk* chunk)
{
    chunk->use_journal_hint = USE_JOURNAL_DEFAULT;
}

static int start_use_journal(uint64_t key, void *data, void *arg, int reference)
{
    struct Chunk* c = (struct Chunk*)data;
    if(reference > 0) {
        c->start_use_journal = true;
    }
    if(reference < 0) {
        LOG_WARN("fdcache key %016x reference is %d", key, reference);
        // c->reset_journal_version = true;
        return 0;
    }
    // reference == 0
    be_chunk_start_use_journal(c);
    return 0;
}

void* sync_start_use_journal(void* arg)
{
    foo_cache_item_iterate_reference(fdcache, start_use_journal, NULL);
    return NULL;
}

int async_start_use_journal()
{
    st_thread_create(sync_start_use_journal, NULL, 0, 0);
    return 0;
}

uint64_t checksum_filesize(int fd)
{
    uint64_t filesize = file_size(fd);
    if(filesize == 0){
        return 0;
    }
    return filesize - SIZE_4K;
}

uint64_t be_chunk_tail4k_offset(struct Chunk* chunk)
{
//    if(RAW_FS_ENABLED(chunk)){
//        return rfs->hdr->chunk_size - SIZE_4K;
//    }
    return checksum_filesize(chunk->fd);
}

uint64_t get_checksum_offset(struct Chunk* chunk)
{
    uint64_t filesize = be_chunk_tail4k_offset(chunk);
    if(filesize == 0){
        return 0;
    }
    if(filesize % (CRC32_BLOCK_SIZE + sizeof(uint32_t)) == 0){
        // have checksum
        return filesize / (CRC32_BLOCK_SIZE + sizeof(uint32_t)) * CRC32_BLOCK_SIZE;
    }
    return filesize + SIZE_4K; // no checksum file, no 4K tail too
}

int be_chunk_data_size(struct Chunk* chunk)
{
    return get_checksum_offset(chunk);
}

static inline uint64_t get_checksum_area_len(uint64_t checksum_offset)
{
    return checksum_offset / CRC32_BLOCK_SIZE * sizeof(uint32_t);
}

void* be_chunk_mmap(struct Chunk* chunk, uint32_t offset_in_chunk, uint32_t len,
        int protect, int flag)
{
//    if(RAW_FS_ENABLED(chunk)){
//        return raw_chunk_mmap(rfs, chunk->index_in_raw, offset_in_chunk, len, protect, flag);
//    }
    return mmap(NULL, len, protect, flag, chunk->fd, offset_in_chunk);
}

static inline void map_checksum(struct Chunk* chunk)
{
    if(!chunk->disk_checksum){
        chunk->checksum_map = NULL;
        return;
    }
    chunk->checksum_len = get_checksum_area_len(chunk->checksum_file_offset);
    chunk->checksum_map = (uint32_t*)be_chunk_mmap(chunk, chunk->checksum_file_offset,
            chunk->checksum_len, PROT_WRITE|PROT_READ, MAP_SHARED);
    if((MAP_FAILED == chunk->checksum_map)){
        LOG_FATAL("error map chunk, volumeid %08x chunkid %d", chunk->volumeid,
                chunk->index);
        chunk->checksum_map = NULL;
    }
    return;
}

int be_chunk_pread(struct Chunk* chunk, void* buf, uint32_t size,
                   uint32_t offset, void* mmap_ptr);
static inline void alloc_checksum(struct Chunk* chunk)
{

    if(!chunk->disk_checksum){
        LOG_DEBUG("chunk not suport checksum");
        return;
    }
    chunk->checksum_len = get_checksum_area_len(chunk->checksum_file_offset);
    chunk->checksum_alloc = (uint32_t*)memalign(SIZE_4K, chunk->checksum_len);
    LOG_INFO("pread file len is %d offset is %d", chunk->checksum_len, chunk->checksum_file_offset);
    int ret = be_chunk_pread(chunk, chunk->checksum_alloc, chunk->checksum_len, chunk->checksum_file_offset, NULL);
    if(ret < (int)chunk->checksum_len){
        LOG_ERROR("error fill checksum");
        free(chunk->checksum_alloc);
        chunk->checksum_alloc = NULL;
        return;
    }
    LOG_DEBUG("success alloc checksum");
}

static inline void chunk_init_checksum(struct Chunk* chunk, bool checksum)
{
    chunk->disk_checksum = checksum;
    chunk->chunk_file_size = be_chunk_tail4k_offset(chunk);
    chunk->tail_offset = chunk->chunk_file_size;
    chunk->checksum_file_offset = be_chunk_data_size(chunk);
#ifndef DIV_CHUNK
#ifndef ALLOC_CHECKSUM
    LOG_INFO("map_checksum volumeid %08x index %d", chunk->volumeid, chunk->index);
    map_checksum(chunk);
#else
    LOG_INFO("alloc_checksum volumeid %08x index %d", chunk->volumeid, chunk->index);
	alloc_checksum(chunk);
#endif
#else
    #ifndef ALLOC_CHECKSUM
	map_whole_file(chunk);
#else
	alloc_checksum_div(chunk);
#endif
#endif
}

void chunk_init(struct Chunk* chunk, uint32_t volumeid, uint32_t index, int fd,
        int index_in_raw, uint64_t version_in_memory, uint64_t version_in_disk,
        bool checksum, bool ec_chunk)
{
    new_a_chunk();

    chunk->fd = fd;
    chunk->version = version_in_memory;
    chunk->version_in_disk = version_in_disk;
    chunk->inflight_write_replicate = 0;
    chunk->volumeid = volumeid;
    chunk->index = index;
    chunk->incubation = NULL;
    chunk->deleted = false;
    chunk->index_in_raw = index_in_raw;
    chunk->being_incubated = false;
    chunk->replicate_write_lock = st_mutex_new();
    range_lock_init(&chunk->lock);
    journal_lite_init(&chunk->journal, version_in_memory);
    chunk_init_checksum(chunk, checksum);
//    chunk_init_ec(chunk, ec_chunk);
//    check_and_init_ec_journal(chunk);
}

__dev_t dev_nb(int fd)
{
    struct stat sb;
    int ret = fstat(fd, &sb);
    if (unlikely(ret < 0)) {
        LOG_ERROR("unable to fstat file %d: %s", errno, strerror(errno));
        return 0;
    }
    if (S_ISCHR(sb.st_mode) || S_ISBLK(sb.st_mode)) {
        return sb.st_rdev;
    }

    return sb.st_dev;
}

#ifndef BLKPBSZGET
#define BLKSSZGET  _IO(0x12,104)/* get block device sector size */
#endif

int dev_sector_size(__dev_t dev_nb)
{
    int maj, min;
    maj = gnu_dev_major(dev_nb);
    min = gnu_dev_minor(dev_nb);
    LOG_INFO("maj:min (%d:%d)", maj, min);
    char dev_path[128];
    sprintf(dev_path, "/dev/block/%d:%d", maj, min);
    int fd = open(dev_path, O_RDONLY);
    if(fd <= 0) {
        LOG_ERROR("error open file %s %m", dev_path);
        return 0;
    }

    int sector_size = 0;
    int ret = ioctl(fd, BLKSSZGET, &sector_size);
    if(ret < 0) {
        LOG_ERROR("error ioctl %m");
        sector_size = 0;
    }
    close(fd);
    return sector_size;
}

int fd_sector_size(int fd)
{
    return dev_sector_size(dev_nb(fd));
}

int file_sector_size(const char* path)
{
    int fd = open(path, O_RDONLY);
    if(fd <= 0) {
        LOG_ERROR("ERROR open file %s %m", path);
        return 0;
    }
    int size = fd_sector_size(fd);
    close(fd);
    return size;
}

void check_disk_align(const char* path)
{
    bool default_available = test_size(path, o_direct_size);
    if(default_available) {
        LOG_INFO("o_direct_size %d is available", o_direct_size);
        return;
    }

    int tmp = o_direct_size;
    o_direct_size = file_sector_size(path);
    if(o_direct_size != 0){
        LOG_INFO("disk sector size is %d this is for o_direct", o_direct_size);
        return;
    }
    LOG_ERROR("g_disk_align is 0, file_sector_size must error, set to default");
    o_direct_size = tmp;
}

struct CheckerArg
{
    const char* ckroot;
    bool stop;
    st_thread_t th;
};

void* disk_broken_checker(void* _arg)
{
    struct CheckerArg *arg = (struct CheckerArg *)_arg;
    char *buf = (char*)ursax_4k_align_alloc(SIZE_4K);
    memset(buf, 0, SIZE_4K);
    char path[PATH_MAX];
    sprintf(path, "%s/check_disk_broken", arg->ckroot);
    int fd = open(path, O_DIRECT | O_CREAT | O_RDWR, 0600);
    if(fd <= 0){
        LOG_ERROR("error open path %s %m", path);
        goto out;
    }
    unlink(path);
    while(!arg->stop){
        int ret = st_sleep(10*60);
//		int ret = st_sleep(10*6); /*zym: 60 -> 6 for debugging st*/
        if(ret < 0){
            LOG_INFO("disk_broken_checker return ahead, continue");
            continue;
        }
        ret = pwrite(fd, buf, SIZE_4K, 0);
        if(ret == SIZE_4K){
            continue;
        }
        LOG_ERROR("error pwrite check ret %d, chunkserver will exit here", ret);
        exit(-1);
    }
    close(fd);
    out:
    free(buf);
    return NULL;
}

void* start_disk_broken_checker(const char* ckroot)
{
    struct CheckerArg* checker = (struct CheckerArg*)zalloc(sizeof(*checker));
    checker->ckroot = ckroot;
    checker->stop = false;
    checker->th = st_thread_create(&disk_broken_checker, checker, 1, 0);
    return checker;
}

int stop_disk_broken_checker(void* data)
{
    struct CheckerArg* checker = (struct CheckerArg*)data;
    checker->stop = true;
    st_thread_interrupt(checker->th);
    st_thread_join(checker->th, NULL);
    return 0;
}



int be_check_xattr(const char *root_path)
{
#ifndef DISABLE_XATTR_CHECK
#define TEST_FILE "test_xattr"
    char filepath[PATH_MAX];
    sprintf(filepath, "%s/" TEST_FILE, root_path);
    int tmpfd = open(filepath, O_RDWR | O_CREAT, 0644);
    if(tmpfd < 0){
        LOG_ERROR("error open file %s %m", filepath);
        return -1;
    }
    char tmp[PATH_MAX];
    int ret = fsetxattr(tmpfd, USER_VERSION, filepath, strlen(filepath)+1, 0);
    LOG_INFO("fsetxattr ret %d", ret);
    int ret1 = fgetxattr(tmpfd, USER_VERSION, tmp, sizeof(tmp));
    LOG_INFO("fgetxattr get %s, ret1 is %d", tmp, ret1);
    close(tmpfd);
    unlink(filepath);
    return ret;
#else
    return 0;
#endif
}

int be_flock()
{
    static char filename[] = "ursax.cs.flock";
    lockfd = open(filename, O_RDWR | O_DIRECT | O_CREAT, 0644);
    if (unlikely(lockfd < 0)) {
        LOG_ERROR("error opening lock file %s %u: %s", filename, errno, strerror(errno));
        return lockfd;
    }

    int ret = flock(lockfd, LOCK_EX | LOCK_NB);
    if (unlikely(ret < 0)) {
        close(lockfd);
        LOG_ERROR("error locking lock file %s %u: %s", filename, errno, strerror(errno));
        return ret;
    }

    return 0;
}

int be_pidfile(bool create)
{
    static char filename[] = "ursax.cs.pid";
    int pidfd = open(filename, O_RDWR | O_CREAT, 0644);
    if (unlikely(pidfd < 0)) {
        LOG_ERROR("error (re)creating pid file %s %u: %s", filename, errno, strerror(errno));
        return ERROR_DISK_IO;
    }

    if (create) {
        char buf[1024];
        int length = snprintf(buf, sizeof(buf), "%32u", getpid());
        int retw = write(pidfd, buf, length);
        int retc = close(pidfd);
        if (unlikely(retw<0 || retc<0)) {
            LOG_ERROR("error writing pid file %s %u: %s", filename, errno, strerror(errno));
            close(pidfd);
            return ERROR_DISK_IO;
        }
    }

    close(pidfd);
    return 0;
}

int be_funlock()
{
    if (-1 == lockfd) {
        return 0;
    }

    int ret = close(lockfd);
    if(unlikely(ret < 0)) {
        LOG_ERROR("error locking lock file %u: %s", errno, strerror(errno));
    }

    lockfd = -1;
    return 0;
}

int read_file(const char* filename, char* buf, int size)
{
    int fd;
    MAKE_FD_ELASTIC((fd = open(filename, O_RDONLY)) >=0, NULL, ERROR_DISK_IO);
    if (unlikely(fd < 0)) {
        LOG_ERROR("failed to open file %s %u: %s", filename, errno, strerror(errno));
        buf[0] = '\0';
        return ERROR_DISK_IO;
    }

    int len = read(fd, buf, size);
    if (unlikely(len <= 0)) {
        LOG_ERROR("failed to read file %s %u: %s", filename, errno, strerror(errno));
        buf[0] = '\0';
        close(fd);
        return ERROR_DISK_IO;
    }

    if (likely(len < size)) {
        buf[ len ] = '\0';
    }
    close(fd);
    return len;
}

int be_check_directory_version()
{
    char buf[1024];
    static char filename[] = "mebs.cs.version";
    int ret = read_file(filename, buf, sizeof(buf));
    if (unlikely(ret < 0)) {
        LOG_ERROR("failed to read version file %s %u: %s", filename, errno, strerror(errno));
        return ERROR_DISK_IO;
    }

    if (unlikely(strcmp(buf, DIRECTORY_VERSION)!=0)) {
        LOG_FATAL("server root directory version doesn't match: supporting '"
        DIRECTORY_VERSION"', got '%s'", buf);
        return ERROR_DISK_IO;
    }

    return 0;
}


static int check_and_open_rfs()
{
    LOG_INFO("open rfs");
    char raw_path[4096];
    char raw_content[4096];
    uint64_t chunksize;
    static char filename[] = USE_RAW_FS_FILE;
    int ret = access(filename, F_OK);
    if(ret < 0) {
        LOG_INFO("error access filename %s %m", filename);
        return 0;
    }

    /* rfs is an unimplemented feature
    ret = read_file(filename, raw_content, sizeof(raw_content));
    if (unlikely(ret < 0)) {
        LOG_ERROR("failed to read version file %s %u: %s", filename, errno, strerror(errno));
        return ERROR_DISK_IO;
    }
    // 遇到 逗号 结束
    // sscanf(raw_content, "%[^,],%u", raw_path, (uint32_t*)&chunksize);
    sscanf(raw_content, "%[^,],%u", raw_path, (unsigned int*)&chunksize);
    LOG_INFO("raw_path:%s, chunksize:%u", raw_path, chunksize);

    rfs = open_raw_chunk_fs(raw_path);
    if(rfs == NULL) {
        return ERROR_DISK_IO;
    }
    */

    return 0;
}

// iterate the items of an dir, NOT recursively
int dir_iterate(const char* dir, dir_iterate_cb callback, void* arg)
{
    DIR *d;
    MAKE_FD_ELASTIC((d = opendir(dir)), NULL, ERROR_DISK_IO);
    if (unlikely(!d))
    {
        LOG_ERROR("error opening dir '%s' %d: %s", dir, errno, strerror(errno));
        return -1;
    }

    int n = 0;
    struct dirent* entry;
    while( (entry = readdir(d)) )
    {
        if (unlikely(entry->d_name[0] == '.'))
            continue;
        if (likely(callback) && callback(entry, arg) < 0)
            break;
        n++;
    }
    closedir(d);
    return n;
}

void free_checksum(struct Chunk* chunk)
{
    if(!chunk->disk_checksum) {
        LOG_DEBUG("this chunk did not support checksum");
        return;
    }
    if(!chunk->checksum_alloc) {
        LOG_WARN("checksum_alloc is NULL, when free!!!");
        return;
    }
    free(chunk->checksum_alloc);
    chunk->checksum_alloc = NULL;
}

void unmap_checksum(struct Chunk* chunk)
{
    if(!chunk->disk_checksum) {
        LOG_DEBUG("this chunk did not support checksum");
        return;
    }
    if(!chunk->checksum_map) {
        LOG_WARN("checksum map is NULL, when unmap!!!!");
        return;
    }
    munmap(chunk->checksum_map, chunk->checksum_len);
    LOG_DEBUG("unmap_checksum %016x success", chunk->key);
}

void chunk_fini_checksum(struct Chunk* c)
{
#ifndef DIV_CHUNK
#ifndef ALLOC_CHECKSUM
    unmap_checksum(c);
#else
    free_checksum(c);
#endif
#else
    #ifndef ALLOC_CHECKSUM
	unmap_whole_file(c);
#else
	free_checksum_div(c);
#endif
#endif
}

bool no_flush_journal = false;
void chunk_journal_flush(struct Chunk* c)
{
    if(!c->deleted && c->journal.version0 != ((uint64_t)-1) && !no_flush_journal)
        // todo check if chunk is set deleted, do not write journal then
        journal_lite_flush(&c->journal);
}

static inline void del_a_chunk()
{
    g_opened_chunk_nb--;
    g_opened_chunk_mem_use -= sizeof(struct Chunk);
    LOG_INFO("del a chunk, opened chunk nb is %d opened chunk mem use is %ld",
            g_opened_chunk_nb, g_opened_chunk_mem_use);
}

void chunk_fini(struct Chunk* c)
{
    if (unlikely(c->lock.locked_range[0][0] != -1
       || c->lock.locked_range[1][0] != -1)) {
        LOG_ERROR("freeing a locked chunk struct");
    }
    st_mutex_destroy(c->replicate_write_lock);
    chunk_fini_checksum(c);
//    fini_chunk_ec_journal(c);
//    chunk_fini_ec(c);
    chunk_journal_flush(c);
    close(c->fd);
    free(c);

    del_a_chunk();
}

void free_cache_entry(void *entry)
{
    struct Chunk* c = (struct Chunk*)entry;
    chunk_fini(c);
}

bool be_check_root_ssd_file()
{
    int ret = access(USE_SSD_FILE, F_OK);
    if(ret == 0) {
        LOG_INFO("use ssd set true");
        return true;
    }

    return false;
}

int be_open_server_root(const char* path)
{
//    struct foo_cache* test = (struct foo_cache*)zalloc(sizeof(struct foo_cache));

//    register_local_log_level("s.backend-fs");

    if (unlikely(path==NULL)) {
        LOG_ERROR("path can't be NULL");
        return -1;
    }

    check_disk_align(path);
    // check_disk_rate(path);
    checker = start_disk_broken_checker(path);

    int ret = chdir(path);
    if (unlikely(ret < 0)) {
        LOG_ERROR("error chdir to '%s' %d: %s", path, errno, strerror(errno));
        return ret;
    }

    ret = be_check_xattr(path);
    if(ret < 0) {
        LOG_ERROR("be_check_xattr ret %d, %m", ret);
        return ret;
    }

    ret = be_flock();
    if (unlikely(ret < 0))
        return -1;

    ret = be_pidfile(true);
    if (unlikely(ret < 0)) {
        be_funlock();
        return -1;
    }

    ret = be_check_directory_version();
    if (unlikely(ret < 0)) {
        be_pidfile(false);
        be_funlock();
        return -1;
    }

    ret = check_and_open_rfs();
    if(ret < 0){
        be_pidfile(false);
        be_funlock();
        return ret;
    }

    open_chunk_lock = st_mutex_new();
    fdcache = foo_cache_create(100 * SIZE_1K, &free_cache_entry, true);

    return 0;
}


int be_chunk_list(uint32_t volumeid, uint32_t index, struct ChunkID** result)
{
    LOG_DEBUG("%s(volumeid=%08x, index=%u)", __func__, volumeid, index);

    uint32_t capacity = 1024*1024 / sizeof(struct ChunkID);
    struct ChunkID* list = (struct ChunkID*)malloc(capacity * sizeof(*list));
    uint32_t size = 0;		// size of the list array

    for (int i=0; i<4096; ++i) {
        char buf[PATH_MAX];
        sprintf_chunk_path_hash(buf, i);

        DIR* d = opendir(buf);
        if (unlikely(d==NULL)) {
            LOG_ERROR("opendir(%s) error %d: %s", buf, errno, strerror(errno));
            continue;
        }

        if (likely(i>0)) {
            st_sleep(0);	// avoid blocking the process too long
        }

        while (true) {
            errno = 0;
            struct dirent* entry = readdir(d);
            if (unlikely(entry == NULL)) {
                if (unlikely(errno != 0))
                    LOG_ERROR("readdir(%s) error %d: %s", buf, errno, strerror(errno));
                break;
            }

            struct ChunkID* x = &list[ size ];
            int ret = sscanf(entry->d_name, "%x.%u", &x->volumeid, &x->index);
            if (unlikely(ret != 2))
                continue;
            if (unlikely(index != 0xffffffff && index != x->index))
                continue;
            if (unlikely(volumeid != 0xffffffff && volumeid != x->volumeid))
                continue;
            if (strchr(entry->d_name, 'j'))// journal file
                continue;
            if(strstr(entry->d_name, ".ecm"))
                continue;
            if(strstr(entry->d_name, ".ecd"))
                continue;
//            if (with_raw())
//            {
//                char chunk_path[512];
//                strcpy(chunk_path, buf);
//                strcat(chunk_path, entry->d_name);
//                check_and_register_raw_chunk(chunk_path);
//            }
            size++;
            if (unlikely(size == capacity))
            {
                void* new_ptr = realloc(list, capacity * 2 * sizeof(*list));
                capacity *= 2;
                list = (struct ChunkID*)new_ptr;
            }
        }
        closedir(d);
    }

    *result = list;
    return size;
}

uint32_t get_fd_number(void)
{
    return foo_cache_entry_count(fdcache);
}

int _be_read_version(int fd, uint64_t* out_version)
{
    LOG_DEBUG("%s(fd=%d)", __func__, fd);

    char version[128];
    int ret = fgetxattr(fd, USER_VERSION, version, sizeof(version));
    if (unlikely(ret < 0 || ret == sizeof(version)))
    {
        LOG_ERROR("chunk version read failed %d: %s", errno, strerror(errno));
        return ERROR_DISK_IO;
    }

    int len = ret;
    version[len] = '\0';
    char* comma = strchr(version, ',');
    if (likely(comma))
        *comma = '\0';

    char* end = NULL;
    long value = strtoul(version, &end, 10);
    if (unlikely(end == NULL || end == version || *end != '\0'))
    {
        LOG_ERROR("parse chunk version error");
        return ERROR_DISK_IO;
    }

    if (likely(comma))
    {
        uint32_t hash = strtol(comma+1, &end, 16);
        if (unlikely(end == NULL || end == comma + 1 || *end != '\0'))
        {
            LOG_ERROR("parse chunk version parity error");
            return ERROR_DISK_IO;
        }
        if (unlikely(hash != int64hash(value)))
        {
            LOG_ERROR("chunk version doesn't match parity");
            return ERROR_DISK_IO;
        }
    }

    *out_version = value;
    return 0;
}

int be_chunk_read_version(struct Chunk* chunk, uint64_t* out_version)
{
    return _be_read_version(chunk->fd, out_version);
}

int align_pread_align_size(int fd, void *buf, uint32_t n, uint64_t offset,
        void *file_map, uint32_t align_check_size)
{
    if(file_map){
        memcpy(buf, (void*)((uint64_t)file_map + offset), n);
        return n;
    }

    if( is_align(offset, align_check_size) && is_align(offset+n, align_check_size) ) {
        return ioengine->pread(fd, buf, n, offset);
    }

    struct ChecksumInfo ckinfo;
    parse_range(offset, n, &ckinfo);
    ALIGIN_MEM(buf_align, SIZE_4K);
    uint32_t cpyed = 0;
    uint64_t align_offset = LOWER_FLOOR(offset, SIZE_4K);
    uint64_t align_end = UPPER_FLOOR(offset + n, SIZE_4K);
    int ret;
    if(ckinfo.pre_len){
        ret = ioengine->pread(fd, buf_align, SIZE_4K, align_offset);
        if(ret < SIZE_4K){
            LOG_ERROR("error align_pread1");
            return -1;
        }
        uint32_t size = MIN(SIZE_4K-ckinfo.pre_len, (int)n);
        memcpy((void*)((uint64_t)buf + cpyed), buf_align + ckinfo.pre_len, size);
        cpyed += size;
    }
    if(ckinfo.mid_len){
        ret = ioengine->pread(fd, (void*)((uint64_t)buf + cpyed), ckinfo.mid_len,
                UPPER_FLOOR(offset, SIZE_4K));
        if(ret < ckinfo.mid_len){
            LOG_ERROR("error align_pread1");
            return -1;
        }
        cpyed += ckinfo.mid_len;
    }
    if(ckinfo.suf_len){
        ret = ioengine->pread(fd, buf_align, SIZE_4K, align_end - SIZE_4K);
        if(ret < SIZE_4K){
            LOG_ERROR("error align_pread2");
            return -1;
        }
        uint32_t size = SIZE_4K - ckinfo.suf_len;
        memcpy((void*)((uint64_t)buf + cpyed), buf_align, size);
        cpyed += size;
    }
    return n;
}

int align_pread(int fd, void *buf, uint32_t n, uint64_t offset, void *file_map)
{
    extern uint32_t o_direct_size;
    return align_pread_align_size(fd, buf, n, offset, file_map, o_direct_size);
}

//#define RAW_FS_ENABLED(chunk) (rfs && chunk->index_in_raw != -1)
typedef int (*align_pio_t)(int fd, void *buf, uint32_t n, uint64_t offset, void *file_map);
int be_chunk_pio(struct Chunk* chunk, void* buf, uint32_t size, uint32_t offset,
        void* mmap_ptr, align_pio_t io)
{
    LOG_DEBUG("rfs %d raw index %ld", rfs, chunk->index_in_raw);
//    if(RAW_FS_ENABLED(chunk)){
//        return raw_chunk_io(rfs, chunk->index_in_raw, buf, offset, size, io);
//    }
    return io(chunk->fd, buf, size, offset, mmap_ptr);
}

int be_chunk_pread(struct Chunk* chunk, void* buf, uint32_t size, uint32_t offset,
        void* mmap_ptr)
{
    return be_chunk_pio(chunk, buf, size, offset, mmap_ptr, align_pread);
}


// 0 no checksum, 1 have checksum
int be_get_tail_flags(struct Chunk* chunk, int* checksum, int* ec_chunk)
{
    uint64_t filesize = be_chunk_tail4k_offset(chunk);// real file size - SIZE_4K
    if(IS_CRC_INVOLVED_OFFSET(filesize)){
        ALIGIN_MEM(buf4k, SIZE_4K);
        ChunkExtra4K *ce = (ChunkExtra4K *)buf4k;
        // don't use st_pio
        int ret = be_chunk_pread(chunk, buf4k, SIZE_4K, filesize, NULL);

        if(ret < 0) {
            LOG_ERROR("error read ChunkExtra4K");
            return ERROR_DISK_IO;
        }
        if(ce->info.crc_magic == CRC_MAGIC) {
            *checksum = 1;
        }
        *ec_chunk = ce->info.ec_chunk;
        LOG_DEBUG("open a none crc chunk which has tail");
        return 1;
    }
    LOG_DEBUG("file size same not have tail");
    return 0;
}

// returns 0, ERROR_CHUNK_NOT_FOUND, ERROR_INTERNAL, ERROR_VERSION
static int _be_chunk_open(uint32_t volumeid, uint32_t index, struct Chunk** out_chunk)
{
    LOG_DEBUG("%s(volumeid=%08x, index=%u)", __func__, volumeid, index);

    struct Chunk* chunk = NULL;
    uint64_t key = chunk_make_key(volumeid, index);
    int ret = foo_cache_get_reference(fdcache, key, &chunk);
    if (likely(ret >= 0 && chunk != NULL)) {
        if (unlikely(chunk->deleted)) {
            foo_cache_adjust_reference(fdcache, key, -1);
            return ERROR_CHUNK_NOT_FOUND;
        }
        LOG_DEBUG("SUCCESS get CHUNK, REFRENCE IS %d", ret);
        if(ret > 20){
            LOG_WARN("chunk %08x.%d reference > 20, ref is %d, "
                     "must be some thing error!!!!!!!!!!!!!!!!", volumeid, index, ret);
            // print_call_stack();
        }
        *out_chunk = chunk;
        return 0;
    }
    LOG_DEBUG("no cache get, open from disk");
    char path[PATH_MAX];
    sprintf_chunk_path_filename(path, volumeid, index);

    ret = access(path, F_OK);
    if (unlikely(ret < 0)) {
        LOG_INFO("chunk (%08x.%d) not found %d: %s", volumeid, index,
                errno, strerror(errno));
        return ERROR_CHUNK_NOT_FOUND;
    }

    int fd;
    MAKE_FD_ELASTIC(
            (fd = open(path, O_RDWR|io_extra_flag))>=0,
            "unable to open chunk", ERROR_INTERNAL);

    LOG_DEBUG("path %s", path);
    int index_in_raw = -1;
//    if (with_raw())
//    {
//
//        ret = get_raw_fs_index_by_fd(fd, &index_in_raw);
//        if(ret < 0){
//            close(fd);
//            return ret;
//        }
//    }
    chunk = (struct Chunk*)zalloc(sizeof(*chunk));
    chunk->fd = fd;
    uint64_t version_in_disk, version_start, version_end;
    ret = be_chunk_read_version(chunk, &version_in_disk);
    if (unlikely(ret < 0)){
        close(fd);
        free(chunk);
        return ERROR_VERSION;
    }

    LOG_INFO("chunk %08x.%d version in disk is %ld", volumeid, index, version_in_disk);
    version_end = version_in_disk;
    ret = get_chunk_version_range_from_journal(get_big_journal(USE_JOURNAL_DEFAULT),
            volumeid, index, &version_start, &version_end);
    if(0 == ret) {
        if(version_start != version_in_disk + 1){
            LOG_INFO("chunk %08x.%d version start != version + 1 %ld %ld %ld", volumeid, index, version_start, version_in_disk, version_end);
        }
    }

    int checksum = 0, ec_chunk = false;
    int flag_ret = be_get_tail_flags(chunk, &checksum, &ec_chunk);
    if(flag_ret < 0){
        close(fd);
        free(chunk);
        return flag_ret; // ERROR_DISK_IO
    }

    chunk_init(chunk, volumeid, index, fd, index_in_raw, version_end, version_in_disk, checksum == 1, ec_chunk);
    extern bool current_journal_broken;
    if(current_journal_broken){
        chunk->use_journal_hint = USE_JOURNAL_DONT;
    }
    be_chunk_update_being_incubated(chunk);
    foo_cache_put_reference(fdcache, chunk->key, chunk, 1); // add to fdcache, init ref 1
    LOG_DEBUG("SUCCESS OPEN CHUNK, INIT REFRENCE IS 1");
    *out_chunk = chunk;
    return 0;
}

int be_chunk_open(uint32_t volumeid, uint32_t index, struct Chunk** out_chunk)
{
    st_mutex_lock(open_chunk_lock);
    int ret = _be_chunk_open(volumeid, index, out_chunk);
    st_mutex_unlock(open_chunk_lock);
    if(ret >= 0){
        PUSH_THREAD_CHUNK_ID(volumeid, index);
    }
    return ret;
}

void be_chunk_close(struct Chunk* chunk)
{
    LOG_DEBUG("%s(volumeid=%08x, index=%u)", __func__, chunk->volumeid, chunk->index);
    if(chunk->write_lock_thread == st_thread_self()){
        if(chunk->master_chunk){
            chunk->inflight_write_replicate--;
            chunk->master_chunk = false;
        }
        in_flight_write--;
        chunk->write_lock_thread = NULL;
    }
    unlock_chunk_write(chunk);

    int ref = foo_cache_put_reference(fdcache, chunk->key, chunk, 0);
    if (likely(ref == 0)) {
//        if(unlikely(chunk->reset_journal_version)){
//            LOG_DEBUG("reset chunk %08x.%d journal version", chunk->volumeid, chunk->index);
//            be_chunk_reset_journal_version(chunk);
//            chunk->reset_journal_version = false;
//        }
//        if(unlikely(chunk->start_use_journal)){
//            be_chunk_start_use_journal(chunk);
//            chunk->start_use_journal = false;
//        }
        if(unlikely(chunk->deleted)){
            LOG_INFO("cache reference is 0, chunk_fini called");
            foo_cache_remove(fdcache, chunk->key, NULL);
        } else if(chunk->close_fd){
            LOG_INFO("cache reference is 0, chunk_fini called1");
            foo_cache_remove(fdcache, chunk->key, NULL);
        }
    }
    POP_THREAD_MAGIC();
}

int check_and_close_rfs()
{
    return 0;
}

int be_close_server_root()
{
    be_pidfile(false);
    be_funlock();
    foo_cache_delete(fdcache, 0);
    st_mutex_destroy(open_chunk_lock);
    check_and_close_rfs();
    stop_disk_broken_checker(checker);
    return 0;
}

static const char* hex_chars = "0123456789abcdef";
static int ursax_sprintf_version(char* buf, uint64_t version, uint32_t hash)
{
    char stack_version[32];
    char stack_hash[16];
    int len_version = 0, len_hash = 0, len_buf = 0;
    do {
        stack_version[len_version++] = hex_chars[version % 10];
        version /= 10;
    }
    while(version);
    do {
        stack_hash[len_hash++] = hex_chars[hash % 16];
        hash /= 16;
    }
    while(hash);
    int i = len_version - 1;
    while(i >= 0) {
        buf[len_buf++] = stack_version[i--];
    }
    buf[len_buf++] = ',';
    i = len_hash - 1;
    while(i >= 0) {
        buf[len_buf++] = stack_hash[i--];
    }
    buf[len_buf] = 0;
    return len_buf;
}

int _be_write_version(int fd, uint64_t version)
{
    LOG_DEBUG("%s(fd=%d, version=%llu)", __func__, fd, version);

    char _version[128];
    uint32_t hash = int64hash(version);
    // int len = snprintf(_version, sizeof(_version), "%llu,%x", (long long unsigned int)version, hash);
    int len = ursax_sprintf_version(_version, version, hash);
    int ret = fsetxattr(fd, USER_VERSION, _version, len + 1, 0);
    if (unlikely(ret < 0))
    {
        LOG_ERROR("chunk version write failed %d: %s", errno, strerror(errno));
        return ERROR_DISK_IO;
    }
    return 0;
}

int be_write_version(struct Chunk* chunk, uint64_t version)
{
    return _be_write_version(chunk->fd, version);
}

int align_pwrite(int fd, void *buf, uint32_t n, uint64_t offset, void *file_map)
{
    if( is_size_align(offset) && is_size_align(offset+n) ) {
        return ioengine->pwrite(fd, buf, n, offset);
    }

    struct ChecksumInfo ckinfo;
    parse_range(offset, n, &ckinfo);
    ALIGIN_MEM(align_buf, SIZE_4K);
    uint32_t cpyed = 0;
    uint64_t align_offset = LOWER_FLOOR(offset, SIZE_4K);
    uint64_t align_end = UPPER_FLOOR(offset + n, SIZE_4K);
    int ret;
    if(ckinfo.pre_len) {
        if(file_map){
            ret = SIZE_4K;
            memcpy(align_buf, (char *)file_map + align_offset, SIZE_4K);
        }
        else {
            ret = ioengine->pread(fd, align_buf, SIZE_4K, align_offset);
        }
        if(ret < SIZE_4K) {
            LOG_ERROR("error align_pwrite1");
            return -1;
        }
        uint32_t size = MIN(SIZE_4K-ckinfo.pre_len, (int)n);
        memcpy(align_buf + ckinfo.pre_len, (void *)((uint64_t)buf + cpyed), size);
        ret = ioengine->pwrite(fd, align_buf, SIZE_4K, align_offset);
        if(ret < SIZE_4K) {
            LOG_ERROR("error align_pwrite2");
            return -1;
        }
        cpyed += size;
    }
    if(ckinfo.mid_len) {
        ret = ioengine->pwrite(fd, (void *)((uint64_t)buf + cpyed), ckinfo.mid_len,
                UPPER_FLOOR(offset, SIZE_4K));
        if(ret < ckinfo.mid_len){
            LOG_ERROR("error align_pwrite3");
            return -1;
        }
        cpyed += ckinfo.mid_len;
    }
    if(ckinfo.suf_len) {
        uint64_t off = align_end - SIZE_4K;
        if(file_map){
            ret = SIZE_4K;
            memcpy(align_buf, (char *)file_map + off, SIZE_4K);
        }
        else {
            ret = ioengine->pread(fd, align_buf, SIZE_4K, off);
        }
        if(ret < SIZE_4K) {
            LOG_ERROR("error align_pwrite4");
            return -1;
        }
        uint32_t size = SIZE_4K - ckinfo.suf_len;
        memcpy(align_buf, (void *)((uint64_t)buf + cpyed), size);
        ret = ioengine->pwrite(fd, align_buf, SIZE_4K, off);
        if(ret < SIZE_4K) {
            LOG_ERROR("error align_pwrite5");
            return -1;
        }
    }
    return n;
}

int be_chunk_pwrite(struct Chunk* chunk, void* buf, uint32_t size, uint32_t offset, void* mmap_ptr)
{
    return be_chunk_pio(chunk, buf, size, offset, mmap_ptr, align_pwrite);
}

static inline void sprintf_date(char* buf)
{
    time_t now = time(NULL);
    struct tm* t = localtime(&now);
    strftime(buf, 256, "%Y-%m-%d", t);
}

int sprintf_today_recycle(char* newpath)
{
    char date[64];
    sprintf_date(date);
    int n = sprintf(newpath, "recycled/%s/", date);
    if(access(newpath, F_OK) < 0) {
        mkdir(newpath, 0777);
        st_sleep(1); // for file inotify, don't delete
    }
    return n;
}

int _be_chunk_delete(uint32_t volumeid, uint32_t index, bool real_delete)
{
    LOG_DEBUG("%s(volumeid=%08x, index=%u)", __func__, volumeid, index);

    struct Chunk* chunk = NULL;
    uint64_t key = chunk_make_key(volumeid, index);
    int ref = foo_cache_adjust_reference(fdcache, key, 0);
    if(0 == ref) {
        foo_cache_remove(fdcache, key, &chunk);
//        flush_ec_journal(chunk);
        chunk_fini(chunk);
    }
    else if(ref > 0) {// if ret > 0, someone opened it
        foo_cache_get_reference(fdcache, key, &chunk);
        foo_cache_adjust_reference(fdcache, key, -1);
        chunk->deleted = true;
//        flush_ec_journal(chunk);
        LOG_DEBUG("%s chunk->deleted = true %x %d", __func__, volumeid, index);
    }// else ref < 0, not opened
//    unlink_ec_journal_file(volumeid, index);

    char path[PATH_MAX], newpath[PATH_MAX];
    sprintf_chunk_path_filename(path, volumeid, index);
    int ret = access(path, F_OK);
    if (unlikely(ret < 0))
    {
        LOG_ERROR("chunk (%08x.%d) not found %d: %s", volumeid, index, errno, strerror(errno));
        return ERROR_CHUNK_NOT_FOUND;
    }

    if(real_delete){
        LOG_INFO("real delete chunk %08x.%d, path %s", volumeid, index, path);
        unlink(path);
        return 0;
    }

    int n = sprintf_today_recycle(newpath);

    sprintf_chunk_filename(newpath + n, volumeid, index);
    ret = access(newpath, F_OK);
    if (unlikely(ret == 0))	// file already exists
    {
        file_rotate(newpath);
    }

    char shadow_newpath[PATH_MAX];
//    if (with_raw()) {
//        n = sprintf_today_shadow_recycle(shadow_newpath);
//
//        sprintf_chunk_filename(shadow_newpath + n, volumeid, index);
//
//        ret = _be_chunk_delete_shadow_link(path, shadow_newpath);
//        if (unlikely(ret < 0))
//        {
//            LOG_ERROR("_be_chunk_delete_shadow_link error %d:%s ", errno, strerror(errno));
//            return ret;
//        }
//    }
    ret = rename(path, newpath);
    if (likely(ret < 0)) {
        LOG_ERROR("failed to delete chunk(%08x.%d) by renaming: %d, %s", volumeid, index, errno, strerror(errno));
//        if (with_raw())
//            unlink(shadow_newpath);
        return ERROR_DISK_IO;
    }

    // try to set mtime to now, so that cs-sitter can use mtime as a time of deletion
    utimes(newpath, NULL);

    LOG_INFO("rename %s to %s success, check access %d", path, newpath, access(path, F_OK));
    return 0;
}

int be_chunk_delete(uint32_t volumeid, uint32_t index)
{
    return _be_chunk_delete(volumeid, index, false);
}

uint32_t data_size_to_crc_chunk_size(uint32_t size)
{
    return size + sizeof(uint32_t) * (size / CRC32_BLOCK_SIZE);
}

uint32_t data_size_to_chunk_size(uint32_t size)
{
    return data_size_to_crc_chunk_size(size) + SIZE_4K;
}

static int allocate_fd(char *path, int fd, int len, bool preallocation)
{
    if (preallocation) {
        int ret = fallocate(fd, 0, 0, len);
        if (unlikely(ret < 0)) {
            LOG_ERROR("error fallocate chunk '%s'", path);
            preallocation = false;
        }
    }

    if (!preallocation) {
        int ret = ftruncate(fd, len);
        if (unlikely(ret < 0)) {
            LOG_ERROR("error ftruncate chunk '%s'", path);
            return ERROR_DISK_IO;
        }
    }
    return 0;
}

int _be_chunk_create(char* path, int *raw_index, uint64_t size/*in byte*/,
        bool preallocation, bool with_checksum, bool ec_chunk)
{
    LOG_DEBUG("%s('%s', size=%u, preallocation=%u)", __func__, path, size, preallocation);
//    if(rfs && rfs->hdr->chunk_size != data_size_to_chunk_size(size)){
//        LOG_ERROR("rfs->hdr->chunk_size is %d and len is %d", rfs->hdr->chunk_size, size);
//        return ERROR_INTERNAL;
//    }

    int fd;
    MAKE_FD_ELASTIC(
            (fd = open(path, O_CREAT | O_EXCL | O_RDWR |io_extra_flag, 0644)) >= 0,
            "unable to open chunk", ERROR_INTERNAL);

    // checksum area always exist
    off_t off_4k = data_size_to_crc_chunk_size(size);
    off_t len = data_size_to_chunk_size(size); // CRC_NB_PER_BLOCK -> 4 byte, SIZE_4K for extra data
    int ret;
//    ret = rfs ? allocate_raw_and_fs_link_fd(raw_index, fd) :
//          allocate_fd(path, fd, len, preallocation);
    ret = allocate_fd(path, fd, len, preallocation);
    if(ret < 0) {
        close(fd);
        unlink(path);
        return ret;
    }

    if(with_checksum || ec_chunk) {
        ALIGIN_MEM(buf4k, SIZE_4K);
        memset(buf4k, 0, SIZE_4K);
        ChunkExtra4K *extra = (ChunkExtra4K *)buf4k;
        if(with_checksum) {
            extra->info.crc_magic = CRC_MAGIC;
        }
//        if(ec_chunk) {
//            extra->info.ec_chunk = ec_chunk;
//        }
//        int ret = rfs ? be_raw_chunk_io(*raw_index, buf4k, SIZE_4K, off_4k, true) :
//                  pwrite(fd, buf4k, SIZE_4K, off_4k);
        int ret = pwrite(fd, buf4k, SIZE_4K, off_4k);
        if(ret < 0) {
            LOG_ERROR("error pwrite flag, '%s'", path);
            close(fd);
            unlink(path);
//            set_rfs_index_unused(rfs, *raw_index);
            return ERROR_DISK_IO;
        }
    }
//    if (with_raw()) {
//        ret = _be_chunk_create_shadow_link(path);
//        if (unlikely(ret < 0))
//        {
//            LOG_ERROR("error create link, path '%s'", path);
//            close(fd);
//            unlink(path);
//            set_rfs_index_unused(rfs, *raw_index);
//            return ERROR_INTERNAL;
//        }
//    }

    return fd;
}

int _be_set_checksum_flag(int fd)
{
    // file size can get whether checksum exist
    return 0;
}

int be_chunk_create(uint32_t volumeid, uint32_t index, uint64_t size/*in byte*/,
        bool preallocation, bool incubating, bool with_checksum, bool ec_chunk,
        struct Chunk** ret_chunk)
{
    LOG_DEBUG("%s(volumeid=%08x, index=%u, size=%u, preallocation=%u, incubating=%u)",
              __func__, volumeid, index, size, preallocation, incubating);
    if(size == 0){
        LOG_ERROR("be_chunk_create size is 0, chunk %08x.%d, return error here", volumeid, index);
        return ERROR_INTERNAL;
    }

    char path[PATH_MAX];
    sprintf_chunk_path_filename(path, volumeid, index);
    LOG_DEBUG("chunk_path_filename: %s", path);

    int ret = access(path, F_OK);
    if (unlikely(ret >= 0))
    {
        LOG_ERROR("chunk (%08x.%d) alread exists", volumeid, index);
        return ERROR_CHUNK_EXISTS;
    }

    uint64_t key = chunk_make_key(volumeid, index);
    // wait for old one to be shut down
    int retry_count = 0;
    while((ret = foo_cache_adjust_reference(fdcache, key, 0)) >= 0){
        // delete one that steal in fdcache, just wait for some time
        LOG_INFO("wait fdcache one to be deleted ref %d, chunk %08x.%d", ret, volumeid, index);
        st_usleep(100000); // 100ms
        if(retry_count++ > 5){
            struct Chunk* local_chunk = NULL;
            foo_cache_remove(fdcache, key, &local_chunk);
            LOG_INFO("chunk %p will be mem leak, deleted %d", local_chunk, local_chunk->deleted);
            break;
        }
    }

    int raw_index = -1;
    int fd = _be_chunk_create(path, &raw_index, size, preallocation, with_checksum, ec_chunk);
    if(fd < 0){
        LOG_ERROR("unable to create new chunk, %m");
        return ERROR_INTERNAL;
    }

    uint64_t version = incubating ? -1 : 0;
    ret = _be_write_version(fd, version);
    if (unlikely(ret < 0)) {
        close(fd);
        be_chunk_delete(volumeid, index);
        return ret;
    }

    _be_set_checksum_flag(fd);

    struct Chunk* chunk = (struct Chunk*)zalloc(sizeof(*chunk));
    chunk_init(chunk, volumeid, index, fd, raw_index, version, version, with_checksum, ec_chunk);
    int inited_reference = 0;
    if(ret_chunk){
        *ret_chunk = chunk;
        inited_reference = 1;
    }
    foo_cache_put_reference(fdcache, chunk->key, chunk, inited_reference); // add to fdcache, init ref 0

    if (unlikely(incubating)) {
        ret = be_chunk_set_being_incubated(chunk);
        if (unlikely(ret < 0)) {/* ignored */}
    }
    return 0;
}

int be_chunk_write_version(struct Chunk* chunk)
{
    if(get_big_journal(chunk->use_journal_hint)){
        return sync_chunk_write_version(get_big_journal(chunk->use_journal_hint), chunk->volumeid, chunk->index, chunk->chunk_data_size, chunk->version);
    }
    int ret = be_write_version(chunk, chunk->version);
    chunk->version_in_disk = chunk->version;
    return ret;
}

int be_chunk_get_reference(uint32_t volumeid, uint32_t index)
{
    uint64_t key = chunk_make_key(volumeid, index);
    return foo_cache_adjust_reference(fdcache, key, 0);
}

int be_chunk_open_for_write(uint32_t volumeid, uint32_t index, struct Chunk** out_chunk, bool master, int errno_for_lock)
{
    int ret = be_chunk_open(volumeid, index, out_chunk);
    if(ret < 0){
        return ret;
    }
    int lret = lock_chunk_write(*out_chunk);
    if(lret < 0){
        be_chunk_close(*out_chunk);
        return errno_for_lock;
    }
    in_flight_write++;
    (*out_chunk)->master_chunk = master;
    if(master){
        (*out_chunk)->inflight_write_replicate++;
    }
    (*out_chunk)->write_lock_thread = st_thread_self();
    return ret;
}

void be_chunk_reset_journal_version(struct Chunk* chunk)
{
    chunk->version = chunk->version_in_disk;
    chunk->use_journal_hint = USE_JOURNAL_DONT;
}

static int wipe_out_journal_version(uint64_t key, void *data, void *arg, int reference)
{
    struct Chunk* c = (struct Chunk*)data;
    if(reference > 0) {
        c->reset_journal_version = true;
        return 0;
    }
    if(reference < 0) {
        LOG_WARN("fdcache key %016x reference is %d", key, reference);
        // c->reset_journal_version = true;
        return 0;
    }
    // reference == 0
    be_chunk_reset_journal_version(c);
    return 0;
}

void* sync_wipe_out_journal_version(void* arg)
{
    foo_cache_item_iterate_reference(fdcache, wipe_out_journal_version, NULL);
    return NULL;
}

int async_wipe_out_journal_version()
{
    st_thread_create(sync_wipe_out_journal_version, NULL, 0, 0);
    return 0;
}

int log_buf(void *buf, int len)
{
    char *_buf = (char*)buf;
    char *p = (char*)malloc(len * 2 + 1);
    p[len * 2] = 0;
    for(int i = 0, j = 0; i < len; i++){
        sprintf(&p[j], "%02x", (unsigned char)_buf[i]);
        j++;
        j++;
    }
    LOG_INFO("BUF %p: %s", buf, p);
    free(p);
    return 0;
}

int write_crc32s_to_disk_nommap(struct Chunk* chunk, uint32_t *crc, uint32_t nb,
        uint64_t offset/*buf offset*/)
{
    LOG_DEBUG("crcs:", log_buf(crc, nb*sizeof(uint32_t)));
    uint64_t crc_start_index = /*buf offset*/offset / CRC32_BLOCK_SIZE;
    uint64_t crc_end_index = crc_start_index + nb - 1;
    LOG_DEBUG("crc_start index is %d, crc end index is %d, buf offset is %d", crc_start_index, crc_end_index, offset);
    uint64_t block_4k_start = crc_start_index / CRC_NB_PER_BLOCK;
    uint64_t block_4k_end = crc_end_index / CRC_NB_PER_BLOCK;
    ALIGIN_MEM(buf, SIZE_4K);
    uint64_t start_in_block = crc_start_index & CRC_NB_PER_BLOCK_MASK;
    uint64_t end_in_block = crc_end_index & CRC_NB_PER_BLOCK_MASK;
    uint64_t start = start_in_block,  end = CRC_NB_PER_BLOCK - 1;
    // include start and end
    LOG_DEBUG("block_4k_start %d block_4k_end %d", block_4k_start, block_4k_end);
    LOG_DEBUG("start_in_block %d end_in_block %d", start_in_block, end_in_block);
    for(uint64_t i = block_4k_start; i <= block_4k_end; i++){
        int retp = be_chunk_pread(chunk, buf, SIZE_4K,
                CRC_BLK_TO_FILE_OFFSET(i, chunk->checksum_file_offset), NULL);
        if(retp < SIZE_4K){
            LOG_FATAL("pread buf error");
            return -1;
        }
        LOG_DEBUG("read buf:", log_buf(buf, SIZE_4K));
        if(i == block_4k_end){
            end = end_in_block;
        }
        LOG_DEBUG("memcpy buf %p + %d, len %d", buf, start * sizeof(uint32_t),
                (end - start + 1) * sizeof(uint32_t));
        memcpy(buf + start * sizeof(uint32_t), crc,
                (end - start + 1) * sizeof(uint32_t));
        LOG_DEBUG("after copy buf:", log_buf(buf, SIZE_4K));
        crc += (end - start + 1);
        retp = be_chunk_pwrite(chunk, buf, SIZE_4K,
                CRC_BLK_TO_FILE_OFFSET(i, chunk->checksum_file_offset), NULL);
        if(retp < SIZE_4K){
            LOG_FATAL("pwrite buf error");
            return -1;
        }
        start = 0;
    }
    return 0;
}

int read_crc32_from_disk_nommap(struct Chunk* chunk, uint32_t *crc,
        uint32_t nb_to_read, uint64_t offset/*buf offset*/)
{
    uint64_t crc_start_index = offset / CRC32_BLOCK_SIZE;
    uint64_t crc_end_index = crc_start_index + nb_to_read - 1;
    uint64_t block_4k_start = crc_start_index / CRC_NB_PER_BLOCK;
    uint64_t block_4k_end = crc_end_index / CRC_NB_PER_BLOCK;
    ALIGIN_MEM(buf, (SIZE_4K));
    uint64_t start_in_block = crc_start_index & CRC_NB_PER_BLOCK_MASK;
    uint64_t end_in_block = crc_end_index & CRC_NB_PER_BLOCK_MASK;
    uint64_t start = start_in_block,  end = CRC_NB_PER_BLOCK - 1;
    LOG_DEBUG("crc_start_index %d crc_end_index %d block_4k_start %d "
              "block_4k_end %d start_in_block %d end_in_block %d",
              crc_start_index, crc_end_index, block_4k_start, block_4k_end,
              start_in_block, end_in_block);
    for(uint64_t i = block_4k_start; i <= block_4k_end; i++){
        int retp = be_chunk_pread(chunk, buf, SIZE_4K,
                CRC_BLK_TO_FILE_OFFSET(i, chunk->checksum_file_offset), NULL);
        if(retp < SIZE_4K){
            LOG_FATAL("pread buf error");
            return -1;
        }
        LOG_DEBUG("read buf:", log_buf(buf, SIZE_4K));
        if(i == block_4k_end){
            end = end_in_block;
        }
        LOG_DEBUG("memcpy buf + %d, len %d", start * sizeof(uint32_t),
                (end - start + 1) * sizeof(uint32_t));
        memcpy(crc, buf + start * sizeof(uint32_t),
                (end - start + 1) * sizeof(uint32_t));
        LOG_DEBUG("cpyed data: ",
                log_buf(crc, (end - start + 1) * sizeof(uint32_t)));
        crc += (end - start + 1);
        start = 0;
    }
    return 0;
}

int crc_write(struct Chunk* chunk, uint32_t *crc, uint32_t nb,
        uint64_t crc_start_index)
{
    uint64_t align_crc_start_index = LOWER_FLOOR(crc_start_index, CRC_NB_PER_BLOCK);
    uint64_t align_crc_end_index =
            UPPER_FLOOR(crc_start_index + nb, CRC_NB_PER_BLOCK);
    uint64_t align_nb = align_crc_end_index - align_crc_start_index;
    uint64_t align_size = align_nb * sizeof(uint32_t);

#ifndef ALLOC_CHECKSUM
    ALIGIN_MEM(align_buf, (SIZE_4K*4));
    memcpy(align_buf, &chunk->checksum_map[align_crc_start_index], align_size);
    LOG_DEBUG("1crc start index %d ali start %d ali end %d ali nb %d",
            crc_start_index, align_crc_start_index, align_crc_end_index, align_nb, log_buf(align_buf, align_size));
    memcpy(align_buf + (crc_start_index - align_crc_start_index)*sizeof(uint32_t),
            crc, nb * sizeof(uint32_t));
    LOG_DEBUG("memcpy offset %d, size %d",
            (crc_start_index - align_crc_start_index)*sizeof(uint32_t),
            nb * sizeof(uint32_t));
    LOG_DEBUG("2crc start index %d ali start %d ali end %d ali nb %d",
            crc_start_index, align_crc_start_index, align_crc_end_index,
            align_nb, log_buf(align_buf, align_size));

    uint64_t align_file_offset = chunk->checksum_file_offset
            + align_crc_start_index * sizeof(uint32_t);
    LOG_DEBUG("pwrite crc32 align file offset %d align size %d",
            align_file_offset, align_size, log_buf(align_buf, align_size));
    int ret = be_chunk_pwrite(chunk, align_buf, align_size, align_file_offset, NULL);
    return ret;
#else
    memcpy(&chunk->checksum_alloc[crc_start_index], crc, nb* sizeof(uint32_t));
	uint64_t align_file_offset = chunk->checksum_file_offset + align_crc_start_index * sizeof(uint32_t);
	int ret = be_chunk_pwrite(chunk, &chunk->checksum_alloc[align_crc_start_index], align_size, align_file_offset, NULL);
	return ret;
#endif
}

int write_crc32s_to_disk(struct Chunk* chunk, uint32_t *crc, uint32_t nb,
        uint64_t offset/*buf offset*/)
{
    if(!chunk->disk_checksum){
        LOG_DEBUG("no checksum chunk writen with checksum");
        return 0;
    }
#ifndef DIRECT_CRC
    if(likely(chunk->checksum_map || chunk->checksum_alloc)) {
        uint64_t crc_start_index = offset / CRC32_BLOCK_SIZE;
#ifdef MEM_SYNC
        memcpy(&chunk->checksum_map[crc_start_index], crc, nb * sizeof(uint32_t));
		flush_checksum(chunk);
#else
        // 1k align
        int ret = crc_write(chunk, crc, nb, crc_start_index);
        if(ret < 0){
            LOG_FATAL("write_crc32s_to_disk error");
            return -1;
        }
#endif
        return 0;
    }
#endif
    LOG_WARN("error chunk->checksum_map is NULL, use none map write");
    return write_crc32s_to_disk_nommap(chunk, crc, nb, offset);
}

int read_crc32_from_disk(struct Chunk* chunk, uint32_t *crc,
        uint32_t nb_to_read, uint64_t offset/*buf offset*/)
{
#ifndef DIRECT_CRC
    if(likely(chunk->checksum_map || chunk->checksum_alloc)){
        uint64_t crc_start_index = offset / CRC32_BLOCK_SIZE;
#ifndef ALLOC_CHECKSUM
        memcpy(crc, &chunk->checksum_map[crc_start_index],
                nb_to_read * sizeof(uint32_t));
#else
        memcpy(crc, &chunk->checksum_alloc[crc_start_index], nb_to_read * sizeof(uint32_t));
#endif
        return 0;
    }
#endif
    LOG_WARN("error chunk->checksum_map is NULL, use none map read");
    return read_crc32_from_disk_nommap(chunk, crc, nb_to_read, offset);
}


// offset is 512 byte aligin
int update_checksum_after_write(struct Chunk* chunk, char *buf, int byte_len,
        uint64_t offset)
{
    if(!chunk->disk_checksum){
        return 0;
    }
    LOG_DEBUG("%s, byte len %d offset %d", __func__, byte_len, offset,
            log_buf(buf, byte_len));
    struct ChecksumInfo ckinfo;
    parse_range(offset, byte_len, &ckinfo);
    uint64_t end = offset + byte_len; // end offset
    uint64_t pre_align_offset = LOWER_FLOOR(offset, CRC32_BLOCK_SIZE);
    uint64_t suf_align_offset = LOWER_FLOOR(end, CRC32_BLOCK_SIZE);
    char *mid_buf = buf + UPPER_FLOOR(offset, CRC32_BLOCK_SIZE) - offset;
    LOG_DEBUG("pre_len %d suf_len %d", ckinfo.pre_len, ckinfo.suf_len);
    LOG_DEBUG("pre_offset %d suf_offset %d mid_len %d", pre_align_offset,
            suf_align_offset, ckinfo.mid_len);
    if(ckinfo.pre_len){
        ALIGIN_MEM(prebuf, CRC32_BLOCK_SIZE);
        LOG_DEBUG("pre len-------------");
#ifndef DIV_CHUNK
        be_chunk_pread(chunk, prebuf, CRC32_BLOCK_SIZE, pre_align_offset, NULL);
#else
        pread_div(chunk, prebuf, CRC32_BLOCK_SIZE, pre_align_offset);
#endif
        LOG_DEBUG("update prelen after read:", log_buf(prebuf, CRC32_BLOCK_SIZE));
        uint32_t crcs[1];
        LOG_DEBUG("crc32_4k_sequence size %d", CRC32_BLOCK_SIZE,
                log_buf(prebuf, CRC32_BLOCK_SIZE));
        crc32_4K_sequence(prebuf, CRC32_BLOCK_SIZE, crcs);
        LOG_DEBUG("crc32 result is %x", crcs[0]);
#ifndef DIV_CHUNK
        int ret = write_crc32s_to_disk(chunk, crcs, 1, pre_align_offset);
#else
        int ret = write_crc32s_to_disk_div(chunk, crcs, 1, pre_align_offset);
#endif
        if(ret < 0){
            LOG_FATAL("write_crc32s_to_disk error1");
            return ret;
        }
    }
    if(ckinfo.suf_len){
        ALIGIN_MEM(sufbuf, CRC32_BLOCK_SIZE);
        LOG_DEBUG("suf len-------------");
#ifndef DIV_CHUNK
        be_chunk_pread(chunk, sufbuf, CRC32_BLOCK_SIZE, suf_align_offset, NULL);
#else
        pread_div(chunk, sufbuf, CRC32_BLOCK_SIZE, suf_align_offset);
#endif
        LOG_DEBUG("update sufbuf after read:", log_buf(sufbuf, CRC32_BLOCK_SIZE));
        uint32_t crcs[1];
        LOG_DEBUG("crc32_4k_sequence size %d", CRC32_BLOCK_SIZE,
                log_buf(sufbuf, CRC32_BLOCK_SIZE));
        crc32_4K_sequence(sufbuf, CRC32_BLOCK_SIZE, crcs);
        LOG_DEBUG("crc32 result is %x", crcs[0]);
#ifndef DIV_CHUNK
        int ret = write_crc32s_to_disk(chunk, crcs, 1, suf_align_offset);
#else
        int ret = write_crc32s_to_disk_div(chunk, crcs, 1, suf_align_offset);
#endif
        if(ret < 0){
            LOG_FATAL("write_crc32s_to_disk error2");
            return ret;
        }
    }
    if(ckinfo.mid_len){
        LOG_DEBUG("mid len-------------");
        uint32_t mid_blk_nb = ckinfo.mid_len / CRC32_BLOCK_SIZE;
        // todo 最大单次写入 4k*4k = 16M, 4k crc ret
        uint32_t crcs[SIZE_4K];
        crc32_4K_sequence(mid_buf, ckinfo.mid_len, crcs);
        LOG_DEBUG("CRC RESULTS are:", log_buf(crcs, mid_blk_nb * sizeof(uint32_t)));
#ifndef DIV_CHUNK
        int ret = write_crc32s_to_disk(chunk, crcs, mid_blk_nb,
                UPPER_FLOOR(offset, CRC32_BLOCK_SIZE));
#else
        int ret = write_crc32s_to_disk_div(chunk, crcs, mid_blk_nb, UPPER_FLOOR(offset, CRC32_BLOCK_SIZE));
#endif
        if(ret < 0){
            LOG_FATAL("write_crc32s_to_disk error3");
            return ret;
        }
    }
    return 0;
}

int be_chunk_raw_info(struct Chunk* chunk, int* fd, off_t *raw_offset, off_t offset)
{
/*    if(RAW_FS_ENABLED(chunk)){
        *fd = rfs->fd;
        *raw_offset = get_real_offset(rfs, chunk->index_in_raw, offset);
        LOG_DEBUG("raw index %d raw offset %d offset", chunk->index_in_raw, *raw_offset, offset);
    }*/
    return 0;
}

ssize_t journaled_st_pread(struct Chunk* chunk, void* buf, size_t count, off_t offset)
{
    if(!get_big_journal(chunk->use_journal_hint)){
        return be_chunk_pread(chunk, buf, count, offset, NULL);
    }
    return sync_chunk_pread(get_big_journal(chunk->use_journal_hint), (be_chunk_pread_t)&be_chunk_pread, chunk, chunk->volumeid, chunk->index, buf, count, offset);
}

int be_chunk_flush(struct Chunk* chunk)
{
    LOG_DEBUG("%s(volumeid=%08x, index=%u)", __func__, chunk->volumeid, chunk->index);

    // int ret = fdatasync(chunk->fd);
    int ret = fsync(chunk->fd);
    if (unlikely(ret < 0))
    {
        LOG_ERROR("be_chunk_flush: fsync() or fdatasync() failed %d: %s", errno, strerror(errno));
        return ERROR_DISK_IO;
    }
    return 0;
}